package net.bwie.realtime.jtp.dws.log.function;

import net.bwie.realtime.jtp.dws.log.bean.PageViewBean;
import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * 创建窗口函数实例，对每分钟窗口中数据进行计算
 */
public class PageViewWindowFunction implements WindowFunction<PageViewBean, String, String, TimeWindow> {

    //格式化实例
    private FastDateFormat format= FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss");
    @Override
    public void apply(String s, TimeWindow timeWindow,
                      Iterable<PageViewBean> iterable, Collector<String> collector) throws Exception {
        //1-获取窗口开始和结束时间
        String windowStartTime = format.format(timeWindow.getStart());
        String windowEndTime = format.format(timeWindow.getEnd());

        //2-对窗口中数据计算
        Long pvCount=0L;
        Long pvDuringTime=0L;
        Long uvCount=0L;
        Long sessionCount=0L;
        for(PageViewBean pageViewBean : iterable){
            pvCount+=pageViewBean.getPvCount();
            pvDuringTime+=pageViewBean.getPvDuringTime();
            uvCount+=pageViewBean.getUvCount();
            sessionCount+=pageViewBean.getSessionCount();
        }

        //3-输出
        String output = windowStartTime + "," + windowEndTime + ","
                +windowStartTime.substring(0,10)+","
                + s + ","
                + uvCount + "," + sessionCount + "," + pvCount + "," + pvDuringTime;

        collector.collect(output);
    }

}
